#include "plugin.h"
#include "commtraff.h"
#include "sqlite3_utils.h"
#include "common_list.h"
#include "common.h"
#include <string.h>
#include <time.h>

typedef struct {
    int32_t  batchInsertThreshold;
    int32_t  recordStorageTime; // 单位为秒
} WriteConfig;

static  WriteConfig g_config = {
        .batchInsertThreshold = 100 ,
        .recordStorageTime = 1 * 24 * 60 * 60, // 默认1天
};

static const char *config_keys[] = {BATCH_INSERT_THRESHOLD_KEY, RECORD_STORAGE_TIME_KEY};

static int config_keys_num = STATIC_ARRAY_SIZE(config_keys);


// 传输层链表头节点，使用接口名、源地址、目的地址作为标识
typedef struct {
    pthread_mutex_t lock;
    unsigned int cnt;
    ListNode list;
} CommTraffList;

// 声明全局链表头指针和互斥锁
static CommTraffList g_transportLists = {
    .lock = PTHREAD_MUTEX_INITIALIZER,
    .cnt = 0,
};
static CommTraffList g_sessionLists = {
    .lock = PTHREAD_MUTEX_INITIALIZER,
    .cnt = 0,
};
static CommTraffList g_rpcLists = {
    .lock = PTHREAD_MUTEX_INITIALIZER,
    .cnt = 0,
};
static CommTraffList g_wpaLists = {
    .lock = PTHREAD_MUTEX_INITIALIZER,
    .cnt = 0,
};

// 添加LinkTraff链表
static CommTraffList g_linkLists = {
    .lock = PTHREAD_MUTEX_INITIALIZER,
    .cnt = 0,
};

static CommTraffList g_hisyseventLists = {
    .lock = PTHREAD_MUTEX_INITIALIZER,
    .cnt = 0,
};

// 声明数据库上下文
static DbContext *g_dbContext = NULL;


// 批量插入函数指针类型定义
typedef int32_t (*BatchInsertFunc)(DbContext *ctx, ListNode *list);

// 通用的数据刷新函数
static int FlushDataCommon(CommTraffList *traffList, BatchInsertFunc insertFunc, const char *dataType)
{
    if (g_dbContext == NULL) {
        ERROR("Database context is not initialized");
        return COMMTRAFF_FAIL;
    }
    if (IsListEmpty(&traffList->list) || traffList->cnt == 0) {
        INFO("No %s data to flush", dataType);
        return COMMTRAFF_SUCCESS;
    }
    // 执行批量插入
    int32_t rc = insertFunc(g_dbContext, &traffList->list);

    // 重置链表状态
    if (rc == COMMTRAFF_SUCCESS) {
        ListNode *item = NULL;
        ListNode *nextItem = NULL;
        CommTraffNode *info = NULL;
        // 释放剩余的链表节点
        LIST_FOR_EACH_SAFE(item, nextItem, &traffList->list) {
            info = LIST_ENTRY(item, CommTraffNode, node);
            ListDelete(&info->node);
            free(info);
        }
        ListInit(&traffList->list);
        traffList->cnt = 0;
        return COMMTRAFF_SUCCESS;
    } else {
        ERROR("Failed to batch insert %s data", dataType);
        return COMMTRAFF_FAIL;
    }
}


// 创建表如果不存在的辅助函数
static int CreateTableIfNotExists(DbContext *ctx, TableNameID tableId)
{
    bool isExist = false;
    const char *tableName = NULL;
    int32_t rc;
    
    // 根据表ID确定表名
    if (tableId == TABLE_SESSION_TRAFF) {
        tableName = "SessionTraff";
    } else if (tableId == TABLE_TRANSPORT_TRAFF) {
        tableName = "TransportTraff";
    } else if (tableId == TABLE_RPC_TRAFF) {
        tableName = "RpcTraff";
    } else if (tableId == TABLE_WPA_TRAFF) {
        tableName = "WpaTraff";
    } else if(tableId == TABLE_LINK_TRAFF) {
        tableName = "LinkTraff";
    } else if(tableId == TABLE_HISYSEVENT_TRAFF) {
        tableName = "HisyseventTraff";
    } else {
        ERROR("Unknown table id: %d", tableId);
        return COMMTRAFF_FAIL;
    }
    
    // 检查表是否存在
    rc = CheckTableExist(ctx, tableId, &isExist);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Failed to check %s table existence", tableName);
        return COMMTRAFF_FAIL;
    }
    
    // 如果表不存在则创建表
    if (!isExist) {
        rc = CreateTable(ctx, tableId);
        if (rc != COMMTRAFF_SUCCESS) {
            ERROR("Failed to create %s table", tableName);
            return COMMTRAFF_FAIL;
        }
        INFO("%s table created successfully", tableName);
    } else {
        INFO("%s table already exists", tableName);
    }
    
    return COMMTRAFF_SUCCESS;
}

static void DoDailyCleanup(DbContext *ctx)
{
    if (ctx == NULL || ctx->db == NULL) {
        ERROR("DoDailyCleanup: invalid database context");
        return;
    }
    time_t nowSec = time(NULL);
    int64_t cutoff = (int64_t)nowSec - (int64_t)g_config.recordStorageTime;
    (void)DeleteOlderThan(ctx, TABLE_SESSION_TRAFF, cutoff);
    (void)DeleteOlderThan(ctx, TABLE_TRANSPORT_TRAFF, cutoff);
    (void)DeleteOlderThan(ctx, TABLE_RPC_TRAFF, cutoff);
    (void)DeleteOlderThan(ctx, TABLE_WPA_TRAFF, cutoff);
    (void)DeleteOlderThan(ctx, TABLE_LINK_TRAFF, cutoff);
    (void)DeleteOlderThan(ctx, TABLE_HISYSEVENT_TRAFF, cutoff);
}

static int CommTraffWriteInit(void)
{
    // 初始化全局链表和互斥锁
    pthread_mutex_init(&g_linkLists.lock, NULL);
    ListInit(&g_linkLists.list);
    g_linkLists.cnt = 0;

    pthread_mutex_init(&g_transportLists.lock, NULL);
    ListInit(&g_transportLists.list);
    g_transportLists.cnt = 0;

    pthread_mutex_init(&g_sessionLists.lock, NULL);
    ListInit(&g_sessionLists.list);
    g_sessionLists.cnt = 0;

    pthread_mutex_init(&g_rpcLists.lock, NULL);
    ListInit(&g_rpcLists.list);
    g_rpcLists.cnt = 0;

    pthread_mutex_init(&g_wpaLists.lock, NULL);
    ListInit(&g_wpaLists.list);
    g_wpaLists.cnt = 0;

    pthread_mutex_init(&g_hisyseventLists.lock, NULL);
    ListInit(&g_hisyseventLists.list);
    g_hisyseventLists.cnt = 0;
    
    // 打开数据库连接
    int32_t rc = OpenDatabase(&g_dbContext);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Failed to open database");
        return COMMTRAFF_FAIL;
    }
    
    // 检查并创建所有需要的表
    rc = CreateTableIfNotExists(g_dbContext, TABLE_SESSION_TRAFF);
    if (rc != COMMTRAFF_SUCCESS) {
        CloseDatabase(g_dbContext);
        g_dbContext = NULL;
        return COMMTRAFF_FAIL;
    }
    
    rc = CreateTableIfNotExists(g_dbContext, TABLE_TRANSPORT_TRAFF);
    if (rc != COMMTRAFF_SUCCESS) {
        CloseDatabase(g_dbContext);
        g_dbContext = NULL;
        return COMMTRAFF_FAIL;
    }
    
    rc = CreateTableIfNotExists(g_dbContext, TABLE_RPC_TRAFF);
    if (rc != COMMTRAFF_SUCCESS) {
        CloseDatabase(g_dbContext);
        g_dbContext = NULL;
        return COMMTRAFF_FAIL;
    }
    
    rc = CreateTableIfNotExists(g_dbContext, TABLE_WPA_TRAFF);
    if (rc != COMMTRAFF_SUCCESS) {
        CloseDatabase(g_dbContext);
        g_dbContext = NULL;
        return COMMTRAFF_FAIL;
    }
    rc = CreateTableIfNotExists(g_dbContext, TABLE_LINK_TRAFF);
    if (rc != COMMTRAFF_SUCCESS) {
        CloseDatabase(g_dbContext);
        g_dbContext = NULL;
        return COMMTRAFF_FAIL;
    }
    rc = CreateTableIfNotExists(g_dbContext, TABLE_HISYSEVENT_TRAFF);
    if (rc != COMMTRAFF_SUCCESS) {
        CloseDatabase(g_dbContext);
        g_dbContext = NULL;
        return COMMTRAFF_FAIL;
    }
    INFO("CommTraffWriteInit init success");
    return COMMTRAFF_SUCCESS;
}

// 刷新传输层数据到数据库
static int FlushTransportData(void)
{
    return FlushDataCommon(&g_transportLists, InsertTransportTraffBatch, "transport");
}

// 刷新会话层数据到数据库
static int FlushSessionData(void)
{
    return FlushDataCommon(&g_sessionLists, InsertSessionTraffBatch, "session");
}

// 刷新RPC层数据到数据库
static int FlushRpcData(void)
{
    return FlushDataCommon(&g_rpcLists, InsertRpcTraffBatch, "rpc");
}

// 刷新WPA层数据到数据库
static int FlushWpaData(void)
{
    return FlushDataCommon(&g_wpaLists, InsertWpaTraffBatch, "wpa");
}

// 刷新Link层数据到数据库
static int FlushLinkData(void)
{
    return FlushDataCommon(&g_linkLists, InsertLinkTraffBatch, "link");
}

static int FlushHisyseventData(void)
{
    return FlushDataCommon(&g_hisyseventLists, InsertHisyseventTraffBatch, "hisysevent");
}

// 提取的传输层数据处理函数
static int ProcessTransportData(const data_set_t *data_set, const value_list_t *value_list)
{
    // 确保数据值数量正确
    if (value_list->values_len < 8) {
        ERROR("Invalid transport data: values_len < 8");
        return COMMTRAFF_INVALID_PARAM;
    }

    // 创建传输层流量数据结构
    CommTraffNode *node = (CommTraffNode *)calloc(1, sizeof(CommTraffNode));
    if (node == NULL) {
        ERROR("Failed to allocate memory for transport traff node");
        return COMMTRAFF_MEM_ALLOC_FAIL;
    }

    TransportTraff *transportTraff = &node->data.data.transportTraff;
    node->data.traffType = TRANSPORT_LAYER_DATA;
    // 填充其他字段
    transportTraff->pid = (uint16_t)value_list->values[0].gauge;
    transportTraff->time = (time_t)value_list->values[1].gauge;
    transportTraff->sockId = (uint32_t)value_list->values[2].gauge;
    transportTraff->protocol = (uint8_t)value_list->values[3].gauge;
    transportTraff->sendBytes = (uint64_t)value_list->values[4].gauge;
    transportTraff->sendPacks = (uint64_t)value_list->values[5].gauge;
    transportTraff->recvBytes = (uint64_t)value_list->values[6].gauge;
    transportTraff->recvPacks = (uint64_t)value_list->values[7].gauge;

    // 从meta数据中提取字符串值
    if (value_list->meta != NULL) {
        char *ifName = NULL;
        char *srcAddr = NULL;
        char *dstAddr = NULL;
        
        // 获取接口名
        if (meta_data_get_string(value_list->meta, "ifName", &ifName) == 0) {
            strncpy((char *)transportTraff->ifName, ifName, IF_NAMESIZE - 1);
            transportTraff->ifName[IF_NAMESIZE - 1] = '\0';
            free(ifName);
        }
        // 获取源地址
        if (meta_data_get_string(value_list->meta, "srcAddr", &srcAddr) == 0) {
            strncpy(transportTraff->local, srcAddr, INET6_ADDRSTRLEN - 1);
            transportTraff->local[INET6_ADDRSTRLEN - 1] = '\0';
            free(srcAddr);
        }
        // 获取目的地址
        if (meta_data_get_string(value_list->meta, "dstAddr", &dstAddr) == 0) {
            strncpy(transportTraff->remote, dstAddr, INET6_ADDRSTRLEN - 1);
            transportTraff->remote[INET6_ADDRSTRLEN - 1] = '\0';
            free(dstAddr);
        }
    }

    ListInit(&node->node);

    // 直接添加到传输层链表，
    pthread_mutex_lock(&g_transportLists.lock);
    ListTailInsert(&g_transportLists.list, &node->node);
    // 增加计数
    g_transportLists.cnt++;
    INFO("Successfully processed transport data for %s-%s-%s, cnt=%d", transportTraff->ifName, transportTraff->local, transportTraff->remote, g_transportLists.cnt);

    // 检查是否达到批量插入阈值
    if (g_transportLists.cnt >= g_config.batchInsertThreshold) {
        // 执行批量插入
        FlushTransportData();
    }
    pthread_mutex_unlock(&g_transportLists.lock);
    return COMMTRAFF_SUCCESS;
}

// 提取的会话层数据处理函数
static int ProcessSessionData(const data_set_t *data_set, const value_list_t *value_list)
{
    // 确保数据值数量正确
    if (value_list->values_len < 9) {
        ERROR("Invalid session data: values_len < 9");
        return COMMTRAFF_INVALID_PARAM;
    }

    // 创建会话层流量数据结构
    CommTraffNode *node = (CommTraffNode *)calloc(1, sizeof(CommTraffNode));
    if (node == NULL) {
        ERROR("Failed to allocate memory for session traff node");
        return COMMTRAFF_MEM_ALLOC_FAIL;
    }

    SessionTraff *sessionTraff = &node->data.data.sessionTraff;
    node->data.traffType = SESSION_LAYER_DATA;
    // 填充其他字段
    sessionTraff->pid = (uint16_t)value_list->values[0].gauge;
    sessionTraff->time = (time_t)value_list->values[1].gauge;
    sessionTraff->handle = (int32_t)value_list->values[2].gauge;
    sessionTraff->channelId = (uint32_t)value_list->values[3].gauge;
    sessionTraff->connId = (uint32_t)value_list->values[4].gauge;
    sessionTraff->sessionId = (uint32_t)value_list->values[5].gauge;
    sessionTraff->sessionType = value_list->values[6].gauge;
    sessionTraff->sendBytes = (uint32_t)value_list->values[7].gauge;
    sessionTraff->recvBytes = (uint32_t)value_list->values[8].gauge;
    // 从meta数据中提取字符串值
    if (value_list->meta != NULL) {
        char *localSessionName = NULL;
        char *peerSessionName = NULL;
        char *peerUdidHash = NULL;
        if (meta_data_get_string(value_list->meta, "localSessionName", &localSessionName) == 0) {
            strncpy((char *)sessionTraff->localSessionName, localSessionName, SESSION_NAME_LEN);
            sessionTraff->localSessionName[SESSION_NAME_LEN - 1] = '\0';
            free(localSessionName);
        }
        if (meta_data_get_string(value_list->meta, "peerSessionName", &peerSessionName) == 0) {
            strncpy((char *)sessionTraff->peerSessionName, peerSessionName, SESSION_NAME_LEN);
            sessionTraff->peerSessionName[SESSION_NAME_LEN - 1] = '\0';
            free(peerSessionName);
        }
        if (meta_data_get_string(value_list->meta, "peerUdidHash", &peerUdidHash) == 0) {
            strncpy((char *)sessionTraff->peerUdidHash, peerUdidHash, COMMTRAFF_UDID_HASH_LEN);
            sessionTraff->peerUdidHash[COMMTRAFF_UDID_HASH_LEN - 1] = '\0';
            free(peerUdidHash);
        }
    }
    ListInit(&node->node);

    // 直接添加到会话层链表，不处理具体索引
    pthread_mutex_lock(&g_sessionLists.lock);
    ListTailInsert(&g_sessionLists.list, &node->node);
    g_sessionLists.cnt++;
    INFO("Successfully processed session data for %s-%s, cnt=%d", sessionTraff->localSessionName, sessionTraff->peerSessionName, g_sessionLists.cnt);
    // 检查是否达到批量插入阈值
    if (g_sessionLists.cnt >= g_config.batchInsertThreshold) {
        // 执行批量插入
        FlushSessionData();
    }
    pthread_mutex_unlock(&g_sessionLists.lock);
    return COMMTRAFF_SUCCESS;
}

// 提取的RPC层数据处理函数
static int ProcessRpcData(const data_set_t *data_set, const value_list_t *value_list)
{
    // 确保数据值数量正确
    if (value_list->values_len < 7) {
        ERROR("Invalid rpc data: values_len < 7");
        return COMMTRAFF_INVALID_PARAM;
    }

    // 创建RPC层流量数据结构
    CommTraffNode *node = (CommTraffNode *)calloc(1, sizeof(CommTraffNode));
    if (node == NULL) {
        ERROR("Failed to allocate memory for rpc traff node");
        return COMMTRAFF_MEM_ALLOC_FAIL;
    }

    RpcTraff *rpcTraff = &node->data.data.rpcTraff;
    node->data.traffType = RPC_LAYER_DATA;
    // 填充字段 - 修复数据类型匹配
    rpcTraff->proxyHandle = (int32_t)value_list->values[0].gauge;
    rpcTraff->stubHandle = (int32_t)value_list->values[1].gauge;
    rpcTraff->sendDataLen = (uint32_t)value_list->values[2].gauge;
    rpcTraff->recvDataLen = (uint32_t)value_list->values[3].gauge;
    rpcTraff->recvDataSeq = (uint64_t)value_list->values[4].gauge;
    rpcTraff->sendDataSeq = (uint64_t)value_list->values[5].gauge;
    rpcTraff->time = (time_t)value_list->values[6].gauge;
    rpcTraff->localpid = (pid_t)value_list->values[7].gauge;
    rpcTraff->localuid = (uid_t)value_list->values[8].gauge;
    rpcTraff->peerpid = (pid_t)value_list->values[9].gauge;
    rpcTraff->peeruid = (uid_t)value_list->values[10].gauge;
    ListInit(&node->node);

    // 直接添加到RPC层链表
    pthread_mutex_lock(&g_rpcLists.lock);
    ListTailInsert(&g_rpcLists.list, &node->node);
    g_rpcLists.cnt++;
    INFO("Successfully processed rpc data for proxyHandle=%d, stubHandle=%d, cnt=%d", rpcTraff->proxyHandle, rpcTraff->stubHandle, g_rpcLists.cnt);
    // 检查是否达到批量插入阈值
    if (g_rpcLists.cnt >= g_config.batchInsertThreshold) {
        // 执行批量插入
        FlushRpcData();
    }
    pthread_mutex_unlock(&g_rpcLists.lock);
    return COMMTRAFF_SUCCESS;
}

// 提取的WPA层数据处理函数
static int ProcessWpaData(const data_set_t *data_set, const value_list_t *value_list)
{
    // 确保数据值数量正确
    if (value_list->values_len < 17) {
        ERROR("Invalid wpa data: values_len < 17");
        return COMMTRAFF_INVALID_PARAM;
    }

    // 创建WPA层流量数据结构
    CommTraffNode *node = (CommTraffNode *)calloc(1, sizeof(CommTraffNode));
    if (node == NULL) {
        ERROR("Failed to allocate memory for wpa traff node");
        return COMMTRAFF_MEM_ALLOC_FAIL;
    }

    WpaTraff *wpaTraff = &node->data.data.wpaTraff;
    node->data.traffType = WPA_LAYER_DATA;
    // 填充字段 - 修复数据类型匹配
    wpaTraff->signal = (int16_t)value_list->values[0].gauge;
    wpaTraff->noise = (int32_t)value_list->values[1].gauge;
    wpaTraff->snr = (uint16_t)value_list->values[2].gauge;
    wpaTraff->txrate = (uint32_t)value_list->values[3].gauge;
    wpaTraff->rxrate = (uint32_t)value_list->values[4].gauge;
    wpaTraff->realTxrate = (uint32_t)value_list->values[5].gauge;
    wpaTraff->realRxrate = (uint32_t)value_list->values[6].gauge;
    wpaTraff->txEfficiency = (uint16_t)value_list->values[7].gauge;
    wpaTraff->frequency = (uint32_t)value_list->values[8].gauge;
    wpaTraff->txPackets = (uint32_t)value_list->values[9].gauge;
    wpaTraff->rxPackets = (uint32_t)value_list->values[10].gauge;
    wpaTraff->chload = (uint32_t)value_list->values[11].gauge;
    wpaTraff->txBytes = (uint64_t)value_list->values[12].gauge;
    wpaTraff->rxBytes = (uint64_t)value_list->values[13].gauge;
    wpaTraff->txFailed = (uint32_t)value_list->values[14].gauge;
    wpaTraff->txRetries = (uint32_t)value_list->values[15].gauge;
    wpaTraff->time = (time_t)value_list->values[16].gauge;

    // 从meta数据中提取字符串值
    if (value_list->meta != NULL) {
        char *ifName = NULL;
        if (meta_data_get_string(value_list->meta, "ifName", &ifName) == 0) {
            strncpy((char *)wpaTraff->ifName, ifName, IF_NAMESIZE - 1);
            wpaTraff->ifName[IF_NAMESIZE - 1] = '\0';
            free(ifName);
        }
    }
    ListInit(&node->node);

    // 直接添加到WPA层链表
    pthread_mutex_lock(&g_wpaLists.lock);
    ListTailInsert(&g_wpaLists.list, &node->node);
    g_wpaLists.cnt++;
    INFO("Successfully processed wpa data for %s -txBytes:%"PRIu64" rxBytes:%"PRIu64" cnt=%d", wpaTraff->ifName, wpaTraff->txBytes, wpaTraff->rxBytes, g_wpaLists.cnt);
    // 检查是否达到批量插入阈值
    if (g_wpaLists.cnt >= g_config.batchInsertThreshold) {
        // 执行批量插入
        FlushWpaData();
    }
    pthread_mutex_unlock(&g_wpaLists.lock);
    return COMMTRAFF_SUCCESS;
}

// 提取的Link层数据处理函数
static int ProcessLinkData(const data_set_t *data_set, const value_list_t *value_list)
{
    // 确保数据值数量正确
    if (value_list->values_len < 8) {
        ERROR("Invalid link data: values_len < 8");
        return COMMTRAFF_INVALID_PARAM;
    }

    // 创建Link层流量数据结构
    CommTraffNode *node = (CommTraffNode *)calloc(1, sizeof(CommTraffNode));
    if (node == NULL) {
        ERROR("Failed to allocate memory for link traff node");
        return COMMTRAFF_MEM_ALLOC_FAIL;
    }

    LinkTraff *linkTraff = &node->data.data.linkTraff;
    node->data.traffType = LINK_LAYER_DATA;
    struct timespec recordTime = CDTIME_T_TO_TIMESPEC(value_list->time);
    linkTraff->time = recordTime.tv_sec;
    linkTraff->rxPackets = (uint64_t)value_list->values[0].derive;
    linkTraff->txPackets = (uint64_t)value_list->values[1].derive;
    linkTraff->rxBytes = (uint64_t)value_list->values[2].derive;
    linkTraff->txBytes = (uint64_t)value_list->values[3].derive;
    linkTraff->rxErrPackets = (uint64_t)value_list->values[4].derive;
    linkTraff->txErrPackets = (uint64_t)value_list->values[5].derive;
    linkTraff->rxDropPackets = (uint64_t)value_list->values[6].derive;
    linkTraff->txDropPackets = (uint64_t)value_list->values[7].derive;
    
    // 从meta数据中提取字符串值
    if (value_list->meta != NULL) {
        char *ifName = NULL;
        if (meta_data_get_string(value_list->meta, "ifName", &ifName) == 0) {
            strncpy((char *)linkTraff->ifName, ifName, IF_NAMESIZE - 1);
            linkTraff->ifName[IF_NAMESIZE - 1] = '\0';
            free(ifName);
        }
    }
    ListInit(&node->node);

    // 直接添加到Link层链表
    pthread_mutex_lock(&g_linkLists.lock);
    ListTailInsert(&g_linkLists.list, &node->node);
    g_linkLists.cnt++;
    INFO("Successfully processed link data for %s: rxPackets=%"PRIu64", txPackets=%"PRIu64", cnt=%d", linkTraff->ifName, linkTraff->rxPackets, linkTraff->txPackets, g_linkLists.cnt);
    // 检查是否达到批量插入阈值
    if (g_linkLists.cnt >= g_config.batchInsertThreshold) {
        // 执行批量插入
        FlushLinkData();
    }
    pthread_mutex_unlock(&g_linkLists.lock);
    return COMMTRAFF_SUCCESS;
}

static int ProcessHisyseventData(const data_set_t *data_set, const value_list_t *value_list)
{
    // 确保数据值数量正确
    if (value_list->values_len <= 0) {
        ERROR("Invalid hisysevent data: values_len < 0");
        return COMMTRAFF_INVALID_PARAM;
    }
    pthread_mutex_lock(&g_hisyseventLists.lock);

    // 创建Link层流量数据结构
    for (int i = 0; i < value_list->values_len; ++i) {
        // 跳过值为0的记录
        if (value_list->values[i].gauge == 0) {
            continue;
        }
        CommTraffNode *node = (CommTraffNode *)calloc(1, sizeof(CommTraffNode));
        if (node == NULL) {
            ERROR("Failed to allocate memory for hisysevent traff node");
            pthread_mutex_unlock(&g_hisyseventLists.lock);
            return COMMTRAFF_MEM_ALLOC_FAIL;
        }

        HisyseventData *hisyseventTraff = &node->data.data.hisyseventData;
        node->data.traffType = HISYSEVENT_DATA;
        // 填充字段
        struct timespec recordTime = CDTIME_T_TO_TIMESPEC(value_list->time);
        hisyseventTraff->time = recordTime.tv_sec;
        hisyseventTraff->value = (uint64_t)value_list->values[i].gauge;

        // 从meta数据中提取字符串值
        if (value_list->meta != NULL) {
            char *domain = NULL;
            if (meta_data_get_string(value_list->meta, "domain", &domain) == 0) {
                strncpy((char *)hisyseventTraff->domain, domain, 32);
                hisyseventTraff->domain[31] = '\0';
                free(domain);
            }
            char *event = NULL;
            if (meta_data_get_string(value_list->meta, "event", &event) == 0) {
                strncpy((char *)hisyseventTraff->event, event, 32);
                hisyseventTraff->event[31] = '\0';
                free(event);
            }
            char *field = NULL;
            char key[10] = {0};
            snprintf(key, sizeof(key), "field%d", i);
            if (meta_data_get_string(value_list->meta, key, &field) == 0) {
                strncpy((char *)hisyseventTraff->field, field, 32);
                hisyseventTraff->field[31] = '\0';
                free(field);
            }
        }
        ListInit(&node->node);

        ListTailInsert(&g_hisyseventLists.list, &node->node);
        g_hisyseventLists.cnt++;
        INFO("Successfully processed hisysevent data for domain: %s, event: %s, field: %s, value: %"PRIu64", cnt=%d",
             hisyseventTraff->domain, hisyseventTraff->event, hisyseventTraff->field, hisyseventTraff->value, g_hisyseventLists.cnt);
        // 检查是否达到批量插入阈值
        if (g_hisyseventLists.cnt >= g_config.batchInsertThreshold) {
            // 执行批量插入
            FlushHisyseventData();
        }
    }
    pthread_mutex_unlock(&g_hisyseventLists.lock);
    return COMMTRAFF_SUCCESS;
}

static int CommTraffWrite(const data_set_t *data_set, const value_list_t *value_list, user_data_t *user_data)
{
    // 检查参数
    if (data_set == NULL || value_list == NULL) {
        ERROR("Invalid parameters: data_set or value_list is NULL");
        return COMMTRAFF_INVALID_PARAM;
    }
    // 处理传输层数据
    if (strcmp(value_list->type, TRANSPORT_DS_NAME) == 0) {
        return ProcessTransportData(data_set, value_list);
    }
    // 处理会话层数据
    else if (strcmp(value_list->type, SESSION_DS_NAME) == 0) {
        return ProcessSessionData(data_set, value_list);
    }
    // 处理RPC层数据
    else if (strcmp(value_list->type, RPC_DS_NAME) == 0) {
        return ProcessRpcData(data_set, value_list);
    }
    // 处理WPA层数据
    else if (strcmp(value_list->type, WPA_DS_NAME) == 0) {
        return ProcessWpaData(data_set, value_list);
    }
    // 处理Link层数据
    else if (strcmp(value_list->type, LINK_DS_NAME) == 0) {
        return ProcessLinkData(data_set, value_list);
    }
    // 处理hisysevent数据
    else if (strcmp(value_list->plugin, HISYSEVENT_DS_NAME) == 0) {
        return ProcessHisyseventData(data_set, value_list);
    }
    // 其他类型的数据暂不处理
    return COMMTRAFF_SUCCESS;
}

static int CommTraffWriteFlush(cdtime_t timeout, const char *identifier, user_data_t *user_data)
{
    pthread_mutex_lock(&g_transportLists.lock);
    int transportResult = FlushTransportData();
    pthread_mutex_unlock(&g_transportLists.lock);

    pthread_mutex_lock(&g_sessionLists.lock);
    int sessionResult = FlushSessionData();
    pthread_mutex_unlock(&g_sessionLists.lock);
    
    pthread_mutex_lock(&g_rpcLists.lock);
    int rpcResult = FlushRpcData();
    pthread_mutex_unlock(&g_rpcLists.lock);
    
    pthread_mutex_lock(&g_wpaLists.lock);
    int wpaResult = FlushWpaData();
    pthread_mutex_unlock(&g_wpaLists.lock);
    
    // 添加Link层数据刷新
    pthread_mutex_lock(&g_linkLists.lock);
    int linkResult = FlushLinkData();
    pthread_mutex_unlock(&g_linkLists.lock);

    // 添加Hisysevent数据刷新
    pthread_mutex_lock(&g_hisyseventLists.lock);
    int hisyseventResult = FlushHisyseventData();
    pthread_mutex_unlock(&g_hisyseventLists.lock);
    
    INFO("CommTraffWriteFlush Result transport=%d session=%d rpc=%d wpa=%d link=%d hisysevent=%d",
            transportResult, sessionResult, rpcResult, wpaResult, linkResult, hisyseventResult);

    // 数据保留策略
    DoDailyCleanup(g_dbContext);

    return COMMTRAFF_SUCCESS;
}

static int CommTraffWriteShutdown()
{
    // 释放传输层链表资源
    pthread_mutex_lock(&g_transportLists.lock);
    // 刷新剩余数据到数据库
    FlushTransportData();
    pthread_mutex_unlock(&g_transportLists.lock);
    
    // 释放会话层链表资源
    pthread_mutex_lock(&g_sessionLists.lock);
    // 刷新剩余数据到数据库
    FlushSessionData();
    pthread_mutex_unlock(&g_sessionLists.lock);
    
    // 释放RPC层链表资源
    pthread_mutex_lock(&g_rpcLists.lock);
    // 刷新剩余数据
    FlushRpcData();
    pthread_mutex_unlock(&g_rpcLists.lock);
    
    // 释放WPA层链表资源
    pthread_mutex_lock(&g_wpaLists.lock);
    // 刷新剩余数据
    FlushWpaData();
    pthread_mutex_unlock(&g_wpaLists.lock);
    
    // 释放Link层链表资源
    pthread_mutex_lock(&g_linkLists.lock);
    // 刷新剩余数据
    FlushLinkData();
    pthread_mutex_unlock(&g_linkLists.lock);

    pthread_mutex_lock(&g_hisyseventLists.lock);
    FlushHisyseventData();
    pthread_mutex_unlock(&g_hisyseventLists.lock);
    
    // 关闭数据库连接
    if (g_dbContext != NULL) {
        CloseDatabase(g_dbContext);
        g_dbContext = NULL;
    }
    INFO("CommTraffWriteShutdown completed");
    return COMMTRAFF_SUCCESS;
}

static int CommTraffWriteConfig(const char *key, const char *val)
{
    if (key == NULL || val == NULL) {
        ERROR("Invalid configuration key or value");
        return COMMTRAFF_INVALID_PARAM;
    }
    // 处理配置项
    if (strcmp(key,BATCH_INSERT_THRESHOLD_KEY) == 0) {
        int threshold = atoi(val);
        if (threshold <= 0) {
            ERROR("Invalid batch insert threshold: %s", val);
            return COMMTRAFF_INVALID_PARAM;
        }
        g_config.batchInsertThreshold = threshold;
        INFO("Set batch insert threshold to %d", g_config.batchInsertThreshold);
    } else if (strcmp(key, RECORD_STORAGE_TIME_KEY) == 0) {
        int storageTime = atoi(val);
        if (storageTime <= 0) {
            ERROR("Invalid record storage time: %s", val);
            return COMMTRAFF_INVALID_PARAM;
        }
        g_config.recordStorageTime = storageTime;
        INFO("Set record storage time to %d seconds", g_config.recordStorageTime);
    } else {
        ERROR("Unknown configuration key: %s", key);
        return COMMTRAFF_INVALID_PARAM;
    }
    return COMMTRAFF_SUCCESS;
}

void module_register(void)
{
    // 注册插件回调函数
    plugin_register_init("commtraff_write", CommTraffWriteInit);
    plugin_register_config("commtraff_write", CommTraffWriteConfig, config_keys, config_keys_num);
    plugin_register_write("commtraff_write", CommTraffWrite, /* user_data = */ NULL);
    plugin_register_flush("commtraff_write", CommTraffWriteFlush, /* user_data = */ NULL);
    plugin_register_shutdown("commtraff_write", CommTraffWriteShutdown);
}